package com.lzwk.app.dwd.ocpx;


import com.lzwk.base.BaseAppV1;
import com.lzwk.beans.ocpx.DwdOrder;
import com.lzwk.beans.ocpx.DwdPayOrder;
import com.lzwk.beans.ocpx.OdsAccountWxuser;
import com.lzwk.beans.ocpx.Wxuser;
import com.lzwk.common.ocpx.OCPXConstant;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

/**
 * @title: DwdLookUpJoinTables
 * @Author Yeaser
 * @Date: 2022/2/15 14:39
 * @Version 1.0
 * Description: TODO
 */
public class DwdLookUpJoinTablesP extends BaseAppV1 {
    public static void main(String[] args) {
        new DwdLookUpJoinTablesP().init(6,"cks","ocpx_new","");
    }

    @Override
    protected void run(StreamExecutionEnvironment env, StreamTableEnvironment tenv) {
        tenv.executeSql("create table mysql_pay_order(" +
                " `id` INT, " +
                " `payee_id` INT, " +
                " `payee_type` STRING, " +
                " `fee` INT, " +
                " `product_type` INT, " +
                " `product_id` INT, " +
                " `source_id` INT, " +
                " `skin_id` INT, " +
                " `wxapp_id` INT, " +
                " `paid` INT, " +
                " `deliver_status` INT, " +
                " `refund_status` INT, " +
                " `create_time` TIMESTAMP(0), " +
                " `pay_time` TIMESTAMP(0), " +
                " `callback_time` TIMESTAMP(0), " +
                " `outer_order_id` STRING, " +
                " `meta` STRING NOT NULL, " +
                " `ip` STRING, " +
                " `payer_id` INT, " +
                " `source` STRING, " +
                " `is_notified` INT, " +
                " `update_time` TIMESTAMP(0), " +
                " `openid` STRING, " +
                " `proc_time` AS PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='" + OCPXConstant.OCPXSLAVEHOST + "', " +
                "   'port'='" + OCPXConstant.OCPXSLAVEPORT + "', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME + "', " +
                "   'password'='" + OCPXConstant.OCPXSLAVEPASSWORD + "', " +
                "   'database-name'='" + OCPXConstant.OCPXSLAVEDBNAME + "', " +
                "   'table-name'='" + OCPXConstant.OCPXJDBCPAYORDER + "', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'server-id'='35081-35090', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '80000' " +
                ")");

        tenv.executeSql("create table ocpx_account_wxuser(" +
                " `id` INT, " +
                " `openid` STRING, " +
                " `unionid` STRING, " +
                " `headimgurl` STRING, " +
                " `sex` INT, " +
                " `nickname` STRING, " +
                " `subscribed` INT, " +
                " `create_time` TIMESTAMP(0), " +
                " `active_ip` STRING, " +
                " `active_time` TIMESTAMP(0), " +
                " `update_time` TIMESTAMP(0), " +
                " `account_id` INT, " +
                " `wxapp_id` INT, " +
                "  `proc_time` AS PROCTIME(), " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='" + OCPXConstant.OCPXSLAVEHOST + "', " +
                "   'port'='" + OCPXConstant.OCPXSLAVEPORT + "', " +
                "   'username'='" + OCPXConstant.OCPXSLAVEUSERNAME + "', " +
                "   'password'='" + OCPXConstant.OCPXSLAVEPASSWORD + "', " +
                "   'database-name'='" + OCPXConstant.OCPXSLAVEDBNAME + "', " +
                "   'table-name'='" + OCPXConstant.OCPXJDBCACCOUNTWXUSER + "', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '80000' " +
                ")");

        Table wxTable = tenv.sqlQuery(" select * from ocpx_account_wxuser ");

        SingleOutputStreamOperator<OdsAccountWxuser> wxuserDS = tenv.toRetractStream(wxTable, OdsAccountWxuser.class)
                .filter(data -> data.f0)
                .map(data -> data.f1)
                .assignTimestampsAndWatermarks(WatermarkStrategy.
                        <OdsAccountWxuser>forBoundedOutOfOrderness(Duration.ofMinutes(10))
                        .withTimestampAssigner((data, ts) -> data.getCreate_time().getTime())
                );


        Table orderTable = tenv.sqlQuery(" select " +
                " ifnull(id,-1)  as id, " +
                " ifnull(payer_id,-1)  as payer_id, " +
                " ifnull(product_id,-1) as product_id, " +
                " ifnull(wk_account_id,-1) as wk_account_id, " +
                " ifnull(parent_activity_id,-1) as parent_activity_id, " +
                " ifnull(src_prod_id,-1) as src_prod_id, " +
                " ifnull(org_id,-1) as org_id,  " +
                " ifnull(ks_aid,'') as ks_aid, " +
                " ifnull(product_name,'') as product_name, " +
                " ifnull(skin_id,-1) as skin_id, " +
                " ifnull(skin_name,'') as skin_name, " +
                " ifnull(source_id,-1)  as source_id, " +
                " ifnull(source_name,'') as source_name, " +
                " ifnull(wxapp_id,-1) as wxapp_id, " +
                " ifnull(wxapp_nickname,'') as wxapp_nickname,  " +
                " ifnull(nickname,'') as nickname, " +
                " ifnull(headimgurl,'') as headimgurl, " +
                " ifnull(phone_num,'') as phone_num, " +
                " ifnull(creative_id,'') as creative_id, " +
                " ifnull(ad_plan_id,'') as ad_plan_id, " +
                " ifnull(ad_group_id,'') as ad_group_id, " + // MP消耗与订单关联必-1) 须做限制
                " ifnull( gdt_group_id,'') as gdt_group_id, " + //--广点通消耗表与订-1) 单表关联
                " ifnull( bd_creative_id,'') as bd_creative_id, " + //--广点通消耗表与订-1) 单表关联
                " ifnull(fee,0) as fee, " +
                " ifnull(paid,0) as paid, " +
                " create_time, " +
                " pay_time, " +
                " ifnull(tag_id,100) as tag_id, " +
                " ifnull(tag_name ,'') as tag_name, " +
                " ifnull(source_platform,'')  as source_platform, " +
                " ifnull(refund_status,0) as refund_status, " +
                " ifnull(meta,'') as meta, " +
                " ifnull(hash_id,'') as hash_id, " +
                " ifnull(trace_id,'') as trace_id, " +
                " ifnull(appid,'') as appid " +
                " from (select  " +
                " id, " +
                " payer_id, " +
                " product_id, " +
                " wk_account_id, " +
                " parent_activity_id, " +
                " src_prod_id, " +
                " org_id, " +
                " ks_aid, " +
                " product_name, " +
                " skin_id, " +
                " skin_name, " +
                " source_id, " +
                " source_name, " +
                " wxapp_id, " +
                " wxapp_nickname, " +
                " nickname, " +
                " headimgurl, " +
                " phone_num, " +
                " creative_id, " +
                " ad_plan_id, " +
                " if(`method`='wx_jsapi', if(wx_aid <> '',wx_aid, ad_group_id),'') AS ad_group_id, " + // MP消耗与订单关联必须做限制
                " ad_group_id AS gdt_group_id, " + //--广点通消耗表与订单表关联
                " bd_creative_id, " +
                " fee, " +
                " paid, " +
                " create_time, " +
                " pay_time, " +
                " tag_id, " +
                " tag_name , " +
                " source_platform, " +
                " refund_status, " +
                " meta, " +
                " hash_id, " +
                " trace_id, " +
                " appid " +
                " from ( " +
                "   select " +
                "   t1.payer_id, " +
                "   t1.product_id," +
                "   ifnull(t5.account_id,-1) as wk_account_id, " +
                "   ifnull(t8.parent_id,-1) as parent_activity_id, " +
                "   t1.src_prod_id, " +
                "   t1.org_id, " +
                "   t1.ks_aid, " +
                "   t1.product_name, " +
                "   t1.skin_id, " +
                "   t1.skin_name, " +
                "   t1.source_id, " +
                "   t1.source_name, " +
                "   t1.wxapp_id, " +
                "   t1.wxapp_nickname, " +
                "   ifnull(if(t2.src='wx',regexp_replace(t2.nickname,'\\n|\\t',''),regexp_replace(t6.nickname,'\\n|\\t','')),'') as nickname, " +
                "   ifnull(if(t2.src='wx',regexp_replace(t2.headimgurl,'\\n|\\t',''),regexp_replace(t6.headimgurl,'\\n|\\t','')),'') as headimgurl," +
                "   ifnull(if(t2.src<>'wx', regexp_replace(t2.openid,'o_',''), regexp_replace(t7.openid,'o_','')),'') as phone_num, " +
                "   t1.creative_id," +
                "   GetJsonString(t1.meta,'ad_plan_id') as ad_plan_id," +
                "   ifnull(if(starts_with(source_platform,'gdt'),t1.gdt_id,regexp_replace(`t4`.`ad_group_id`,'\\n|\\t','')),'') as ad_group_id, " +
                "   t1.bd_creative_id, " +
                "   t1.fee, " +
                "   t1.paid, " +
                "   t1.create_time, " +
                "   t1.pay_time, " +
                "   t1.wx_aid, " +
                "   t1.`method`, " +
                "   t1.id, " +
                "   t1.tag_id, " +
                "   t1.tag_name, " +
                "   t1.source_platform, " +
                "   t1.refund_status, " +
                "   ifnull(t1.hash_id,'') as hash_id, " +
                "   ifnull(t4.trace_id,'') as trace_id, " +
                "   ifnull(t9.appid,'') as appid, " +
                "   ifnull(`t2`.`parent_account_id`, `t2`.`id`) as order_account_id, " +
                "   t1.meta " +
                "   from " +
                "       (select " +
                "            tt1.`payer_id`, " +
                "            tt1.`product_id`, " +
                "            ifnull(tt1.`source_id`, 1) AS `source_id`, " +
                "            ifnull(tt1.`skin_id`, -1) AS `skin_id`, " +
                "            ifnull(tt1.`wxapp_id`, -1) AS `wxapp_id`, " +
                "            ifnull(tt7.`org_id`, -1) AS `org_id`, " +
                "            ifnull(tt5.`tag_id`, -1) AS `tag_id`, " +
                "            ifnull(regexp_replace(tt3.`name`,'\\n|\\t',''), '') AS `source_name`, " +
                "            ifnull(regexp_replace(tt4.`name`,'\\n|\\t',''), '') AS `skin_name`, " +
                "            ifnull(regexp_replace(tt5.`name`,'\\n|\\t',''), '') AS `product_name`, " +
                "            ifnull(regexp_replace(tt8.`tag_name`,'\\n|\\t',''), '') AS `tag_name`, " +
                "            ifnull(regexp_replace(tt6.`nickname`,'\\n|\\t',''), '') AS `wxapp_nickname`, " +
                "            ifnull(tt5.`src_prod_id`,-1) as src_prod_id, " +
                "            ifnull(CAST(REGEXP_EXTRACT(GetJsonString(tt1.`meta`, 'url'), 'creativeid=([0-9]+)', 1) AS STRING),'') AS creative_id, " +
                "            ifnull(CAST(REGEXP_EXTRACT(GetJsonString(tt1.`meta`, 'url'), 'wx_aid=([0-9]+)', 1) AS STRING),'') AS wx_aid, " +
                "            ifnull(CAST(REGEXP_EXTRACT(GetJsonString(tt1.`meta`, 'url'), 'ks_aid=([0-9]+)', 1) AS STRING)  ,'') AS ks_aid, " +
                "            ifnull(CAST(REGEXP_EXTRACT(GetJsonString(tt1.`meta`, 'url'), 'creative=([0-9]+)', 1) AS STRING)  ,'') AS bd_creative_id, " +
                "            ifnull(GetJsonString(tt1.`meta`, 'openid'),'') AS `open_id`, " +
                "            if(tt1.source='dyshop',GetJsonInt(tt1.`meta`,'pay_amount',0),tt1.`fee`) AS `fee`, " +
                "            ifnull(GetJsonString(tt1.`meta`, 'method'),'') AS `method`, " +
                "            tt1.`payee_id`, " +
                "            ifnull(regexp_replace(tt1.`payee_type`,'\\n|\\t',''),'') AS `payee_type`, " +
                "            ifnull(tt1.`product_type`,-1) AS product_type, " +
                "            tt1.`paid`, " +
                "            tt1.`deliver_status`, " +
                "            tt1.`refund_status`, " +
                "            ifnull(tt1.`create_time`,TO_TIMESTAMP('1970-01-01 08:00:01')) AS `create_time`, " +
                "            ifnull(tt1.`pay_time`,TO_TIMESTAMP('1970-01-01 08:00:01')) AS `pay_time`, " +
                "            ifnull(tt1.`callback_time`,TO_TIMESTAMP('1970-01-01 08:00:01'))  AS `callback_time`, " +
                "            ifnull(regexp_replace(tt1.`outer_order_id`,'\\n|\\t',''),'') AS `outer_order_id`, " +
                "            ifnull(regexp_replace(tt1.`ip`,'\\n|\\t',''),'') AS `ip`, " +
                "            ifnull(regexp_replace(tt1.`source`,'\\n|\\t',''),'') AS `source`, " +
                "            ifnull(tt1.`is_notified`,-1) AS `is_notified`, " +
                "            ifnull(tt1.`update_time`, TO_TIMESTAMP('1970-01-01 08:00:01')) AS `update_time`, " +
                "            ifnull(regexp_replace(tt1.`openid`,'\\n|\\t',''),'') AS `openid`, " +
                "            ifnull(regexp_replace(tt1.`meta`,'\\n|\\t',''),'') AS `meta`, " +
                "            tt1.`id`, " +
                "            ifnull(tt3.`meta`,'') AS `platform_meta`, " +
                "            ifnull(GetJsonString(tt3.`meta`,'source_platform'),'') AS `source_platform`, " +
                "            ifnull(tt3.hash_id,'') AS `hash_id`, " +
                "            tt1.`proc_time`, " +
                "            case when starts_with(ifnull(GetJsonString(tt3.`meta`,'source_platform'),''),'gdt') " +
                "                 then if(ifnull(regexp_extract(GetJsonString(ifnull(regexp_replace(tt1.`meta`,'\\n|\\t',''),''),'url'),'wx_aid=([0-9]+)',1),'')='', " +
                "                     regexp_extract(GetJsonString(ifnull(regexp_replace(tt1.`meta`,'\\n|\\t',''),''),'url'),'adgroupid=([0-9]+)',1), " +
                "                     regexp_extract(GetJsonString(ifnull(regexp_replace(tt1.`meta`,'\\n|\\t',''),''),'url'),'wx_aid=([0-9]+)',1)) " +
                "                else '' " +
                "            end as gdt_id " +
                "            FROM mysql_pay_order   tt1 " +
                "            left join ocpx_promote_promotesource FOR SYSTEM_TIME AS OF `tt1`.`proc_time` as tt3 " +
                "            on tt1.source_id=tt3.id " +
                "            left join ocpx_product_productskin FOR SYSTEM_TIME AS OF `tt1`.`proc_time` as tt4 " +
                "            on tt1.skin_id=tt4.id " +
                "            left join ocpx_product_product FOR SYSTEM_TIME AS OF `tt1`.`proc_time` as tt5 " +
                "            on tt1.product_id = tt5.id " +
                "            left join ocpx_wxapp_wxapp FOR SYSTEM_TIME AS OF `tt1`.`proc_time` tt6 " +
                "            on tt1.wxapp_id = tt6.id " +
                "            left join ocpx_public_admin_recordorgref FOR SYSTEM_TIME AS OF `tt1`.`proc_time` as tt7 " +
                "            on tt1.product_id=tt7.object_id " +
                "            and tt7.object_type = 'product' " +
                "            left join ocpx_tag_info FOR SYSTEM_TIME AS OF `tt1`.`proc_time` as tt8 " +
                "            on tt5.tag_id=tt8.id " +
                "            where tt5.src_prod_type = 'wk_act'   " +
                " ) as t1 " +
                "       left join ods_ocpx_account_account  FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t2 " +
                "           on t1.payer_id = t2.id " +
                "       left join ods_ocpx_wxapp_mpadsubfeedbackdata FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t4 " +
                "           ON t1.open_id = t4.openid" +
                "           AND CAST(UNIX_TIMESTAMP(DATE_FORMAT(t1.pay_time,'yyyy-MM-dd HH:mm:ss')) AS STRING) > t4.subscribe_time  " +
                "       left join adb_account_mapping FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t5 " +
                "           on CAST(t1.payer_id AS STRING) = t5.object_id  " +
                "           and t5.object_type = 'ocpx' " +
                "       left join ods_ocpx_account_account FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t6 " + //得到父id的昵称和头像(wx进来直接取,非wx进来的需要取到绑定的父id的微信信息)
                "           on t2.parent_account_id = t6.id " +
                "       left join ods_ocpx_account_account FOR SYSTEM_TIME AS OF `t1`.`proc_time` as  t7 " + //得到电话号码(非wx进来的需要取到他)
                "           on t2.id = t7.parent_account_id " +
                "       left join adb_division_activity FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t8 " +
                "           on t1.src_prod_id=t8.id " +
                "       left join ocpx_wxapp_wxapp FOR SYSTEM_TIME AS OF `t1`.`proc_time` as t9 " +
                "           on regexp_replace(`t4`.`wechat_account_id`,'\\n|\\t','') = `t9`.`appid` " +
                "           ) t ) m  ");

        SingleOutputStreamOperator<DwdOrder> orderDS = tenv.toRetractStream(orderTable, DwdOrder.class)
                .filter(data -> data.f0)
                .map(data -> data.f1)
                .assignTimestampsAndWatermarks(WatermarkStrategy.
                        <DwdOrder>forBoundedOutOfOrderness(Duration.ofMinutes(10))
                        .withTimestampAssigner((data, ts) -> data.getCreate_time().getTime())
                );

        orderDS.coGroup(wxuserDS)
                .where((KeySelector<DwdOrder, Tuple2<Integer, Integer>>) value -> new Tuple2<>(value.getOrder_account_id(), value.getWxapp_id()))
                .equalTo((KeySelector<OdsAccountWxuser,Tuple2<Integer, Integer>>) value -> new Tuple2<Integer, Integer>(value.getAccount_id(),value.getWxapp_id()))
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .apply(new CoGroupFunction<DwdOrder, OdsAccountWxuser, DwdPayOrder>() {
                    @Override
                    public void coGroup(Iterable<DwdOrder> first, Iterable<OdsAccountWxuser> second, Collector<DwdPayOrder> out) throws Exception {
                        for (DwdOrder dwdOrder : first) {
                            boolean flag = false;
                            //右表中有记录
                            for (OdsAccountWxuser odsAccountWxuser : second) {
                                out.collect(new DwdPayOrder(dwdOrder,odsAccountWxuser));
                                flag = true;
                            }
                            if(!flag){
                                //右表没有记录
                                out.collect(new DwdPayOrder(dwdOrder,true));
                            }
                        }
                    }
                });



    }
}

